package org.example;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.RichProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.RichProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.bean.WaterSensor;
import org.example.partition.MyPartitioner;
import org.example.watermarks.CustomPeriodicWatermarkExample;

import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

public class NCDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(2);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//        env.getConfig().setAutoWatermarkInterval(10000L);
        SingleOutputStreamOperator<WaterSensor> source = env.socketTextStream("localhost", 7777).map(new RichMapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                WaterSensor waterSensor = new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                return waterSensor;
            }
        });

//单调递增，默认升序的水印
//        WatermarkStrategy<WaterSensor> waterSensorWatermarkStrategy = WatermarkStrategy
//                .<WaterSensor>forMonotonousTimestamps()
//                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
//                    @Override
//                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
//                        return element.getTs()*1000;
//                    }
//                });
        //乱序，
        WatermarkStrategy<WaterSensor> waterSensorWatermarkStrategy = WatermarkStrategy
                .<WaterSensor>forBoundedOutOfOrderness(Duration.of(0, ChronoUnit.SECONDS))
                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                    @Override
                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                        return element.getTs()*1000;
                    }
                });

//        WatermarkStrategy<WaterSensor> waterSensorWatermarkStrategy = WatermarkStrategy
//                .<WaterSensor>forGenerator(new WatermarkGeneratorSupplier<WaterSensor>() {
//                    @Override
//                    public WatermarkGenerator<WaterSensor> createWatermarkGenerator(Context context) {
//                        return new CustomPeriodicWatermarkExample(0L);
//                    }
//                })
//                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
//                    @Override
//                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
//                        return element.getTs()*1000;
//                    }
//                });
 source.assignTimestampsAndWatermarks(waterSensorWatermarkStrategy)

//        DataStream<WaterSensor> waterSensorDataStream = waterSensorSingleOutputStreamOperator
//                .partitionCustom(new MyPartitioner(),x->x);


                .map(x-> x.getTs())
                .keyBy(x -> x% 2)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))

//                .reduce(new ReduceFunction<Integer>() {
//                            @Override
//                            public Integer reduce(Integer value1, Integer value2) throws Exception {
//                                return value1 +
//                                        value2;
//                            }
//                        }, new RichProcessAllWindowFunction<Integer, Object, TimeWindow>() {
//
//                            @Override
//                            public void process(ProcessAllWindowFunction<Integer, Object, TimeWindow>.Context context, Iterable<Integer> elements, Collector<Object> out) throws Exception {
//                                context.window().getStart();
//                                DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
//                                LocalDateTime localDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(context.window().getStart()), ZoneId.of("Asia/Shanghai"));// 指定北京时区)
//                                localDateTime.format(formatter);
//                                System.out.println(localDateTime.toString());
//                                out.collect(elements.iterator().next());
//                            }
//
//                            @Override
//                            public void close() throws Exception {
//                                System.out.println("close");
//                                super.close();
//                            }
//
//                            @Override
//                            public void open(Configuration parameters) throws Exception {
//                                System.out.println("open");
//                                super.open(parameters);
//                            }
//                        }
//                ).print();
//        .process(new RichProcessWindowFunction<Integer, Integer, TimeWindow>() {
//            @Override
//            public void process(Object o, ProcessWindowFunction.Context context, Iterable elements, Collector out) throws Exception {
//                DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
//                LocalDateTime localDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(context.window().getStart()), ZoneId.of("Asia/Shanghai"));// 指定北京时区)
//                localDateTime.format(formatter);
//                System.out.println(localDateTime.toString());
//                AtomicReference<Integer> sum = new AtomicReference<>(0);
//                elements.forEach(x->{
//                    sum.updateAndGet(v -> v + x);
//
//                });
//                out.collect(sum.get());
//            }
//
////            @Override
////            public void process(ProcessAllWindowFunction<Integer, Integer, TimeWindow>.Context context, Iterable<Integer> elements, Collector<Integer> out) throws Exception {
////                DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
////                                LocalDateTime localDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(context.window().getStart()), ZoneId.of("Asia/Shanghai"));// 指定北京时区)
////                                localDateTime.format(formatter);
////                                System.out.println(localDateTime.toString());
////                                AtomicReference<Integer> sum = new AtomicReference<>(0);
////                elements.forEach(x->{
////                    sum.updateAndGet(v -> v + x);
////
////                });
////                out.collect(sum.get());
////            }
//
//            @Override
//            public void close() throws Exception {
//                super.close();
//            }
//
//            @Override
//            public void open(Configuration parameters) throws Exception {
//                super.open(parameters);
//            }
//        }).print();
                .process(new ProcessWindowFunction<Long, String, Long, TimeWindow>() {
                    @Override
                    public void process(Long integer, ProcessWindowFunction<Long, String, Long, TimeWindow>.Context context, Iterable<Long> elements, Collector<String> out) throws Exception {
                        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                        LocalDateTime localDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(context.window().getStart()), ZoneId.of("Asia/Shanghai"));// 指定北京时区)
                        localDateTime.format(formatter);
                        System.out.println(localDateTime.toString());
                        out.collect("key:" + integer + "窗口:" + localDateTime.toString() + "数据:" + elements.toString());
                    }
                }

    ).print();
        env.execute();

    }
}
